package com.sure0000.basicKafka.example;

import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

/**
 * @desc
 * @Author xuyouchang
 * @Date 2021/10/21
 */
public class KafkaJsonProducer {
    public static void main(String[] args) {
        // kafka 连接信息
        String topic = "xyc_test_json_02";
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "172.17.1.103:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");



        KafkaProducer<String, String> producer = new KafkaProducer<>(props);


        for (int j = 1; j < 100 ; j++) {
            System.out.println(j);
            // 测试数据
            List<JSONObject> values = new ArrayList<>();
            for (int i = j * 1000000; i < (j+1)*1000000; i++) {
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("id", i);
                jsonObject.put("name", "xyc"+i);
                jsonObject.put("gender", i % 2 == 0 ? "man" : "women");
                values.add(jsonObject);
            }


            for (JSONObject value: values) {
                ProducerRecord<String, String> record = new ProducerRecord<>(topic, null, value.toJSONString());
                producer.send(record);
            }
        }

        producer.close();
    }
}
